#include "destor.h"
#include "jcr.h"
#include "backup.h"
#include "utils/lru_cache.h"
#include "storage/containerstore.h"
#include <openssl/sha.h>
#include <openssl/md5.h>
#include "xdelta/xdelta3.h"

typedef unsigned long long int UINT64;

struct SFRecord {
    fingerprint fp;
    struct superF* superfeature;
    containerid id;
};

struct index_item {
    fingerprint fp;
    containerid id;
};

struct super_feature_item {
    uint64_t sf;
    struct index_item* p_item;
};

struct base_chunk_pointer {
	fingerprint fp;
	containerid id;
};

struct superFeature {
    fingerprint fp;
    uint64_t superF;
    containerid id;
};

GHashTable *delta_sf1;
GHashTable *delta_sf2;
GHashTable *delta_sf3;

GHashTable *delta_hash;

struct lruCache * base_cache;

static pthread_mutex_t mutex;

static pthread_t matching_t;

void init_delta_index(){
    base_cache = new_lru_cache(destor.base_container_num, free_container,
			lookup_fingerprint_in_container);
    
    //delta_index = g_hash_table_new_full(g_int64_hash, g_fingerprint_equal, NULL, free);
    
    delta_sf1 = g_hash_table_new_full(g_int64_hash, g_superfeature_equal, free, NULL);
    delta_sf2 = g_hash_table_new_full(g_int64_hash, g_superfeature_equal, free, NULL);
    delta_sf3 = g_hash_table_new_full(g_int64_hash, g_superfeature_equal, free, NULL);

    delta_hash = g_hash_table_new_full(g_int64_hash, g_fingerprint_equal, NULL, free);

	sds indexpath = sdsdup(destor.working_directory);
	indexpath = sdscat(indexpath, "index/delta.index");
    
    sds sf1_path = sdsdup(destor.working_directory);
	sf1_path = sdscat(sf1_path, "index/SF1");

    FILE *fp_sf1;
    if ((fp_sf1 = fopen(sf1_path, "r"))) {
		/* The number of features */
		UINT64 key_num;
		fread(&key_num, sizeof(UINT64), 1, fp_sf1);
		for (; key_num > 0; key_num--) {
			/* Read a feature */
			struct superFeature* p = (struct superFeature*)
					malloc(sizeof(struct superFeature));
			fread(p->fp, sizeof(fingerprint), 1, fp_sf1);
			fread(&p->superF, sizeof(uint64_t), 1, fp_sf1);
            fread(&p->id, sizeof(containerid), 1, fp_sf1);

            struct super_feature_item* item = (struct super_feature_item*)
					malloc(sizeof(struct super_feature_item));
            
            item->sf = p->superF;
            struct index_item* p_item = g_hash_table_lookup(delta_hash, &p->fp);
            if(!p_item) {
                p_item = (struct index_item*)malloc(sizeof(struct index_item));
                memcpy(p_item->fp, p->fp, sizeof(fingerprint));
                p_item->id = p->id;

                g_hash_table_insert(delta_hash, &p_item->fp, p_item);
            }
            
            item->p_item = p_item;
			g_hash_table_insert(delta_sf1, &item->sf, item);

            free(p);
		}
		fclose(fp_sf1);
	}

    sds sf2_path = sdsdup(destor.working_directory);
	sf2_path = sdscat(sf2_path, "index/SF2");

    FILE *fp_sf2;
    if ((fp_sf2 = fopen(sf2_path, "r"))) {
		/* The number of features */
		UINT64 key_num;
		fread(&key_num, sizeof(UINT64), 1, fp_sf2);
		for (; key_num > 0; key_num--) {
			/* Read a feature */
			struct superFeature* p = (struct superFeature*)
					malloc(sizeof(struct superFeature));
			fread(p->fp, sizeof(fingerprint), 1, fp_sf2);
			fread(&p->superF, sizeof(uint64_t), 1, fp_sf2);
            fread(&p->id, sizeof(containerid), 1, fp_sf2);
            
			struct super_feature_item* item = (struct super_feature_item*)
					malloc(sizeof(struct super_feature_item));
            
            item->sf = p->superF;
            struct index_item* p_item = g_hash_table_lookup(delta_hash, &p->fp);
            if(!p_item) {
                p_item = (struct index_item*)malloc(sizeof(struct index_item));
                memcpy(p_item->fp, p->fp, sizeof(fingerprint));
                p_item->id = p->id;

                g_hash_table_insert(delta_hash, &p_item->fp, p_item);
            }
            
            item->p_item = p_item;
			g_hash_table_insert(delta_sf2, &item->sf, item);

            free(p);
		}
		fclose(fp_sf2);
	}

    sds sf3_path = sdsdup(destor.working_directory);
	sf3_path = sdscat(sf3_path, "index/SF3");

    FILE *fp_sf3;
    if ((fp_sf3 = fopen(sf3_path, "r"))) {
		/* The number of features */
		UINT64 key_num;
		fread(&key_num, sizeof(UINT64), 1, fp_sf3);
		for (; key_num > 0; key_num--) {
			/* Read a feature */
			struct superFeature* p = (struct superFeature*)
					malloc(sizeof(struct superFeature));
			fread(p->fp, sizeof(fingerprint), 1, fp_sf3);
			fread(&p->superF, sizeof(uint64_t), 1, fp_sf3);
            fread(&p->id, sizeof(containerid), 1, fp_sf3);

			struct super_feature_item* item = (struct super_feature_item*)
					malloc(sizeof(struct super_feature_item));
            
            item->sf = p->superF;
            struct index_item* p_item = g_hash_table_lookup(delta_hash, &p->fp);
            if(!p_item) {
                p_item = (struct index_item*)malloc(sizeof(struct index_item));
                memcpy(p_item->fp, p->fp, sizeof(fingerprint));
                p_item->id = p->id;

                g_hash_table_insert(delta_hash, &p_item->fp, p_item);
            }
            
            item->p_item = p_item;
			g_hash_table_insert(delta_sf3, &item->sf, item);

            free(p);
		}
		fclose(fp_sf3);
	}

	sdsfree(indexpath);
    sdsfree(sf1_path);
    sdsfree(sf2_path);
    sdsfree(sf3_path);
    
    if (pthread_mutex_init(&mutex, 0)) {
   		puts("Failed to init delta mutex!");
    	exit(1);
   	}
}

void super_feature_index_update(GQueue *chunks){
    //NOTICE("Filter phase: update %d features", g_hash_table_size(features));
    UINT64 qlen = g_queue_get_length(chunks), i;

    for(i = 0; i < qlen; i++){
        struct chunk* ck = g_queue_peek_nth(chunks, i);
        
        if (ck->deltaCompressed)
            continue;

        struct SFRecord* sfr = (struct SFRecord*)malloc(sizeof(struct SFRecord));
        sfr->superfeature = (struct superF*)malloc(sizeof(struct superF));
    
        memcpy(sfr->fp, ck->fp, sizeof(fingerprint));
        sfr->id = ck->id;
    
        sfr->superfeature->sf1 = ck->superfeature->sf1;
        sfr->superfeature->sf2 = ck->superfeature->sf2;
        sfr->superfeature->sf3 = ck->superfeature->sf3;


        struct index_item* p_item = g_hash_table_lookup(delta_hash, &ck->fp);
        if(!p_item) {
            p_item = (struct index_item*)malloc(sizeof(struct index_item));
            memcpy(p_item->fp, ck->fp, sizeof(fingerprint));
            p_item->id = ck->id;

            g_hash_table_insert(delta_hash, &p_item->fp, p_item);
        }

        struct super_feature_item* item_1 = (struct super_feature_item*)
					malloc(sizeof(struct super_feature_item));
        item_1->sf= sfr->superfeature->sf1;
        item_1->p_item = p_item;
    
        struct super_feature_item* item_2 = (struct super_feature_item*)
					malloc(sizeof(struct super_feature_item));
        item_2->sf = sfr->superfeature->sf2;
        item_2->p_item = p_item;

        struct super_feature_item* item_3 = (struct super_feature_item*)
					malloc(sizeof(struct super_feature_item));
        item_3->sf = sfr->superfeature->sf3;
        item_3->p_item = p_item;

	    pthread_mutex_lock(&mutex);
        g_hash_table_replace(delta_sf1, &item_1->sf, item_1);
        g_hash_table_replace(delta_sf2, &item_2->sf, item_2);
        g_hash_table_replace(delta_sf3, &item_3->sf, item_3);
	    pthread_mutex_unlock(&mutex);
        
        free(sfr->superfeature);
        free(sfr);
    }
}

void close_delta_index(){
    
    jcr.total_sf_num = g_hash_table_size(delta_hash);
    
	sds indexpath = sdsdup(destor.working_directory);
	indexpath = sdscat(indexpath, "index/delta.index");

    sds sf1_path = sdsdup(destor.working_directory);
	sf1_path = sdscat(sf1_path, "index/SF1");

    sds sf2_path = sdsdup(destor.working_directory);
	sf2_path = sdscat(sf2_path, "index/SF2");

    sds sf3_path = sdsdup(destor.working_directory);
	sf3_path = sdscat(sf3_path, "index/SF3");
    
    FILE *fp_sf1;
	if ((fp_sf1 = fopen(sf1_path, "w")) == NULL) {
		perror("Can not open delta.index for write because:");
		exit(1);
	}

    FILE *fp_sf2;
	if ((fp_sf2 = fopen(sf2_path, "w")) == NULL) {
		perror("Can not open delta.index for write because:");
		exit(1);
	}

    FILE *fp_sf3;
	if ((fp_sf3 = fopen(sf3_path, "w")) == NULL) {
		perror("Can not open delta.index for write because:");
		exit(1);
	}

    GHashTableIter iter;
	gpointer key, value;
    
    UINT64 key_num = g_hash_table_size(delta_sf1);
	fwrite(&key_num, sizeof(UINT64), 1, fp_sf1);
	g_hash_table_iter_init(&iter, delta_sf1);
	while (g_hash_table_iter_next(&iter, &key, &value)) {
		struct super_feature_item* p = value;
		fwrite(p->p_item->fp, sizeof(fingerprint), 1, fp_sf1);
		fwrite(&p->sf, sizeof(uint64_t), 1, fp_sf1);
        fwrite(&p->p_item->id, sizeof(containerid), 1, fp_sf1);
	}
	fclose(fp_sf1);

    key_num = g_hash_table_size(delta_sf2);
	fwrite(&key_num, sizeof(UINT64), 1, fp_sf2);
	g_hash_table_iter_init(&iter, delta_sf2);
	while (g_hash_table_iter_next(&iter, &key, &value)) {
		struct super_feature_item* p = value;
		fwrite(p->p_item->fp, sizeof(fingerprint), 1, fp_sf2);
		fwrite(&p->sf, sizeof(uint64_t), 1, fp_sf2);
        fwrite(&p->p_item->id, sizeof(containerid), 1, fp_sf2);
	}
	fclose(fp_sf2);

    key_num = g_hash_table_size(delta_sf3);
	fwrite(&key_num, sizeof(UINT64), 1, fp_sf3);
	g_hash_table_iter_init(&iter, delta_sf3);
	while (g_hash_table_iter_next(&iter, &key, &value)) {
		struct super_feature_item* p = value;
		fwrite(p->p_item->fp, sizeof(fingerprint), 1, fp_sf3);
		fwrite(&p->sf, sizeof(uint64_t), 1, fp_sf3);
        fwrite(&p->p_item->id, sizeof(containerid), 1, fp_sf3);
	}
	fclose(fp_sf3);

    sdsfree(indexpath);
    sdsfree(sf1_path);
    sdsfree(sf2_path);
    sdsfree(sf3_path);

    jcr.sketch_index_item_num += g_hash_table_size(delta_sf1);
    jcr.sketch_index_item_num += g_hash_table_size(delta_sf2);
    jcr.sketch_index_item_num += g_hash_table_size(delta_sf3);
    
	//g_hash_table_destroy(delta_index);
    g_hash_table_destroy(delta_sf1);
    g_hash_table_destroy(delta_sf2);
    g_hash_table_destroy(delta_sf3);

    g_hash_table_destroy(delta_hash);
}

void clear_delta_index() {
    g_hash_table_remove_all(delta_sf1);
    g_hash_table_remove_all(delta_sf2);
    g_hash_table_remove_all(delta_sf3);
}

struct chunk* retrive_basechunk_by_fp_id(fingerprint* fp, containerid id)
{
    struct container *con = NULL;
    struct chunk* c = NULL;
    int32_t res;

    /*
    unsigned char buf[41];
    hash2code(*fp, buf);
    buf[40] = 0;
    if (buf[0] == 'D' && buf[1] == '7' && buf[2] == '8' && buf[3] == '6' && buf[4] == '0' && buf[5] == '7') {
        jcr.zero_chunk_num++;
        printf("fp: %s\n", buf);
    }
    */

	/* first, we find it in the buffer */
	con = retrieve_container_in_write_buffer(id);

    /* then, find it in base cache */
	if (!con) {
        res = check_restore_container(base_cache, fp, id);
    
        if (!res) {
            VERBOSE("Restore cache: container %lld is missed", id);
	        con = retrieve_container_by_id(id);
		    lru_cache_insert(base_cache, con, NULL, NULL);

            jcr.base_cache_miss++;
            jcr.base_container_read++;
        }
        else {
            con = lru_cache_lookup(base_cache, fp);
            jcr.base_cache_hit++;
        }
	}
    else {
        lru_cache_insert(base_cache, con, NULL, NULL);
        con = lru_cache_lookup(base_cache, fp);
        jcr.base_cache_hit++;
    }

    assert(con);
    c = get_basechunk_in_container(con, fp);
    
    assert(c);
    return c;
}

struct chunk* retrive_basechunk_incache_by_fp_id(fingerprint* fp, containerid id)
{
    struct container *con = lru_cache_lookup_without_update(base_cache, fp);
    assert(con);
    
    struct chunk* c = get_basechunk_in_container(con, fp);
    
    assert(c);
    return c;
}

struct chunk* get_simi_chunk(struct chunk* srcChunk) {

    struct super_feature_item* psSuperF = NULL;
	struct super_feature_item* psSuperF1 = NULL;
	struct super_feature_item* psSuperF2 = NULL;
	struct super_feature_item* psSuperF3 = NULL;
    struct chunk* baseChunk = NULL;
	
	/*
    psSuperF = g_hash_table_lookup(delta_sf1, &srcChunk->superfeature->sf1);
    if (!psSuperF) {
        psSuperF = g_hash_table_lookup(delta_sf2, &srcChunk->superfeature->sf2);
        if (!psSuperF) {
            psSuperF = g_hash_table_lookup(delta_sf3, &srcChunk->superfeature->sf3);
        }
    }
    */
    
    psSuperF1 = g_hash_table_lookup(delta_sf1, &srcChunk->superfeature->sf1);
	psSuperF2 = g_hash_table_lookup(delta_sf2, &srcChunk->superfeature->sf2);
	psSuperF3 = g_hash_table_lookup(delta_sf3, &srcChunk->superfeature->sf3);
	
	
    if (psSuperF1)
		psSuperF = psSuperF1;
	else if (psSuperF2)
		psSuperF = psSuperF2;
	else if (psSuperF3)
		psSuperF = psSuperF3;

    if (destor.enable_bestmatch) {
		if (destor.enable_sort_features) {
			if (psSuperF1 && psSuperF2 && !memcmp(psSuperF1->p_item->fp, psSuperF2->p_item->fp, destor.index_key_size))
				psSuperF = psSuperF1;
			else if (psSuperF1 && psSuperF3 && !memcmp(psSuperF1->p_item->fp, psSuperF3->p_item->fp, destor.index_key_size))
				psSuperF = psSuperF1;
			else if (psSuperF2 && psSuperF3 && !memcmp(psSuperF2->p_item->fp, psSuperF3->p_item->fp, destor.index_key_size))
				psSuperF = psSuperF2;
		}
    }
	
	if ((psSuperF1 && psSuperF2)
		|| (psSuperF1 && psSuperF3)
		|| (psSuperF2 && psSuperF3)) {
		jcr.multi_matched_chunk_number++;
	}

	if ((psSuperF1 && psSuperF2 && memcmp(psSuperF1->p_item->fp, psSuperF2->p_item->fp, destor.index_key_size))
		|| (psSuperF1 && psSuperF3 && memcmp(psSuperF1->p_item->fp, psSuperF3->p_item->fp, destor.index_key_size))
		|| (psSuperF2 && psSuperF3 && memcmp(psSuperF2->p_item->fp, psSuperF3->p_item->fp, destor.index_key_size))) {
		jcr.multi_matched_diff_number++;
	}
    
    if (psSuperF) {
        //if(psSuperF->p_item->id <= jcr.container_num_now_written)
        baseChunk = retrive_basechunk_by_fp_id(&psSuperF->p_item->fp, psSuperF->p_item->id);
		jcr.sf_matched_number++;
    }
    
    return baseChunk;

    /*
    struct super_feature_item* psSuperF = NULL;
    struct chunk* baseChunk = NULL;
    struct chunk* baseChunk1 = NULL;
    struct chunk* baseChunk2 = NULL;
    struct chunk* baseChunk3 = NULL;

    if (srcChunk->nosf)
        return baseChunk;
    
    psSuperF = g_hash_table_lookup(delta_sf1, &srcChunk->superfeature->sf1);
    if (psSuperF)
        baseChunk1 = retrive_basechunk_by_fp_id(&psSuperF->p_item->fp, psSuperF->p_item->id);
        
    psSuperF = g_hash_table_lookup(delta_sf2, &srcChunk->superfeature->sf2);
    if (psSuperF)
        baseChunk2 = retrive_basechunk_by_fp_id(&psSuperF->p_item->fp, psSuperF->p_item->id);

    psSuperF = g_hash_table_lookup(delta_sf3, &srcChunk->superfeature->sf3);
    if (psSuperF)
        baseChunk3 = retrive_basechunk_by_fp_id(&psSuperF->p_item->fp, psSuperF->p_item->id);

    int chunkdif1 = -1;
    int chunkdif2 = -1;
    int chunkdif3 = -1;
    
    if (baseChunk1)
        chunkdif1 = baseChunk1->size > srcChunk->size ? 
                    (baseChunk1->size - srcChunk->size)/baseChunk1->size :
                    (srcChunk->size - baseChunk1->size)/baseChunk1->size;

    if (baseChunk2)
        chunkdif2 = baseChunk2->size > srcChunk->size ? 
                    (baseChunk2->size - srcChunk->size)/baseChunk2->size :
                    (srcChunk->size - baseChunk2->size)/baseChunk1->size;

    if (baseChunk3)
        chunkdif3 = baseChunk3->size > srcChunk->size ? 
                    (baseChunk3->size - srcChunk->size)/baseChunk3->size :
                    (srcChunk->size - baseChunk3->size)/baseChunk1->size;

    if (chunkdif1 > 0)
        baseChunk = baseChunk1;
    if (chunkdif2 > 0)
        if (baseChunk) {
            if (chunkdif1 > chunkdif2)
                baseChunk = baseChunk2;
        }
        else
            baseChunk = baseChunk2;
    if (chunkdif3 > 0)
        if (baseChunk)
            if (baseChunk == baseChunk1) {
                if (chunkdif1 > chunkdif3)
                    baseChunk = baseChunk3;
            }
            else if (baseChunk == baseChunk2) {
                if (chunkdif2 > chunkdif3)
                    baseChunk = baseChunk3;
            }
        else
            baseChunk = baseChunk3;

    if (baseChunk = baseChunk1) {
        if (baseChunk2)
            free_chunk(baseChunk2);
        if (baseChunk3)
            free_chunk(baseChunk3);
    }
    else if (baseChunk = baseChunk2) {
        if (baseChunk1)
            free_chunk(baseChunk1);
        if (baseChunk3)
            free_chunk(baseChunk3);
    }
    else if (baseChunk = baseChunk3) {
        if (baseChunk1)
            free_chunk(baseChunk1);
        if (baseChunk2)
            free_chunk(baseChunk2);
    }
                
    return baseChunk;
    */
}

static void* matching_thread(void *arg) {

	while (1) {
        struct chunk* c = sync_queue_pop(resemblance_queue);

        if (c == NULL)
            /* backup job finish */
            break;

        if(CHECK_CHUNK(c, CHUNK_FILE_START) ||
        		CHECK_CHUNK(c, CHUNK_FILE_END) ||
        		CHECK_CHUNK(c, CHUNK_SEGMENT_START) ||
        		CHECK_CHUNK(c, CHUNK_SEGMENT_END)){
        	sync_queue_push(matching_queue, c);
        	continue;
        }

        if(CHECK_CHUNK(c, CHUNK_DUPLICATE)){
        	sync_queue_push(matching_queue, c);
        	continue;
        }

        struct chunk* baseChunk = NULL;
        if (c->size > 2048)
            baseChunk = get_simi_chunk(c);

		int sizediff;
		if (baseChunk) {
			if (baseChunk->size > c->size)
				sizediff = (baseChunk->size - c->size) / (1.0 * baseChunk->size);
			else
				sizediff = (c->size - baseChunk->size) / (1.0 * c->size);
			if (destor.super_feature_method == SUPER_FEATURE) {
				if (sizediff > 0.05) {
					free(baseChunk);
					baseChunk = NULL;
				}
			}
		}
            
        if (baseChunk == NULL) {
            c->delta = NULL;
            c->baseChunk = NULL;
        }
        else
            c->baseChunk = baseChunk;

        sync_queue_push(matching_queue, c);
    }

    sync_queue_term(matching_queue);
	return NULL;
}

void start_matching_phase() {

    init_delta_index();

	matching_queue = sync_queue_new(1000);
	pthread_create(&matching_t, NULL, matching_thread, NULL);
}

void stop_matching_phase() {
	pthread_join(matching_t, NULL);
}
